/**
 * ZooKeeperUtil.java
 */
package com.mht2017.common.utils;

import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

import com.mht2017.common.dto.ProgramStatus;
import org.I0Itec.zkclient.ZkClient;
import org.apache.commons.lang.StringUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.ZooKeeper.States;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;

/**
 * zookeeper的工具类，需要预先设定zkConnectString，通过方法 {@link #setZkConnectString(String)}
 *
 * @author yutong.xiao
 * @date 2017年3月10日 上午11:30:51
 */
public class ZooKeeperUtil {

	private static final Logger LOGGER = LoggerFactory.getLogger(ZooKeeperUtil.class);
	private static final String actionName = "ZooKeeper工具";
	private static final String ZK_PATH_SEPARATOR = "/";
	private static final int MAX_RETRY = 3;

	private static String zkConnectString;
	private static ZooKeeper zooKeeper;
	private static ZkClient zkClient;

	public static boolean setZkConnectString(String zkConnectString) {
		if (StringUtils.isEmpty(ZooKeeperUtil.zkConnectString)) {
			ZooKeeperUtil.zkConnectString = zkConnectString;
			return true;
		} else {
			return false;
		}
	}

	public static ZooKeeper getZooKeeper() {
		checkConnect();
		return zooKeeper;
	}

	public synchronized static ZkClient getZkClient() {
		if (zkClient == null) {
			zkClient = new ZkClient(zkConnectString, 30000, 30000);
			zkClient.setZkSerializer(new ZKStringSerialize());
		}
		return zkClient;
	}

	public synchronized static void setZooKeeper(String zkConnectString) {
		// 如果连接地址和之前的连接不是同一连接地址，则刷新重新连接
		if (zkConnectString != null && StringUtils.isEmpty(ZooKeeperUtil.zkConnectString)
				|| zkConnectString != null && !StringUtils.isEmpty(ZooKeeperUtil.zkConnectString) && zkConnectString != ZooKeeperUtil.zkConnectString) {
			ZooKeeperUtil.zkConnectString = zkConnectString;
			cleanAndCreateConnect();
			KafkaUtil.createOrRefreshConsumer();
		}
	}

	public static boolean exists(String path) throws KeeperException, InterruptedException {
		if (StringUtils.isEmpty(path)) {
			return false;
		}

		checkConnect();
		return null != zooKeeper.exists(path, false);
	}

	/**
	 * 递归删除 因为zookeeper只允许删除叶子节点，如果要删除非叶子节点，只能使用递归
	 *
	 * @param path
	 * @throws IOException
	 */
	public static void rmr(String path, int version) throws Exception {
		ZooKeeper zk = getZooKeeper();
		// 获取路径下的节点
		List<String> children = zk.getChildren(path, false);
		for (String pathCd : children) {
			// 获取父节点下面的子节点路径
			String newPath = "";
			// 递归调用,判断是否是根节点
			if (path.equals("/")) {
				newPath = "/" + pathCd;
			} else {
				newPath = path + "/" + pathCd;
			}
			rmr(newPath, version);
		}
		// 删除节点,并过滤zookeeper节点和 /节点
		if (path != null && !path.trim().startsWith("/zookeeper") && !path.trim().equals("/")) {
			zk.delete(path, version);
		}
	}

	public static String getData(String path) throws KeeperException, InterruptedException {
		byte[] dataRaw = getDataRaw(path);
		if (null != dataRaw) {
			return new String(dataRaw);
		}
		return null;
	}

	public static byte[] getDataRaw(String path) throws KeeperException, InterruptedException {
		if (StringUtils.isEmpty(path)) {
			return new byte[0];
		}

		checkConnect();

		Stat stat = new Stat();
		if (null == zooKeeper.exists(path, false)) {
			LOGGER.warn(String.format("%s-从zk(%s)中获取数据: '%s'，但是节点不存在", actionName, zkConnectString, path));
		} else {
			byte[] data = zooKeeper.getData(path, false, stat);
			if (LOGGER.isDebugEnabled()) {
				LOGGER.debug(String.format("%s-从zk(%s)中获取数据('%s'='(%s)%s', version: '%s')", actionName, zkConnectString, path, SizeUtil.size(data),
						new String(data, 0, Math.min(100, SizeUtil.sizeZeroIfNull(data))), stat.getVersion()));
			}
			return data;
		}
		return null;
	}

	public static int increment(String path) throws KeeperException, InterruptedException {
		if (StringUtils.isEmpty(path)) {
			return -1;
		}

		checkConnect();

		Stat stat = new Stat();
		if (null == zooKeeper.exists(path, false)) {
			LOGGER.warn(String.format("%s-从zk(%s)中获取数据准备自增: '%s'，但是节点不存在", actionName, zkConnectString, path));
		} else {
			int i = increment(path, stat);

			if (i == -1) {
				LOGGER.error("自增失败，请检查节点值是否存在，且支持自增操作（为int）");
				return -1;
			}
			return i;

		}
		return -1;
	}

	private static int increment(String path, Stat stat) {
		int i = Integer.MIN_VALUE;
		try {
			byte[] data = zooKeeper.getData(path, false, stat);
			if (null != data) {
				i = Integer.parseInt(new String(data));
				if (LOGGER.isDebugEnabled()) {
					LOGGER.debug(String.format("%s-从zk(%s)中获取数据('%s'='%s', version: '%s')", actionName, zkConnectString, path, i, stat.getVersion()));
				}
				zooKeeper.setData(path, String.valueOf(i + 1).getBytes(), stat.getVersion());
				return i;
			}
		} catch (Exception e) {
			if (i == Integer.MIN_VALUE) {
				LOGGER.error("请检查路径节点值是否为整数，否则不支持自增操作！");
				return -1;
			}
			try {
				Thread.sleep(100);
			} catch (InterruptedException e1) {
				e1.printStackTrace();
			}
			return increment(path, stat);
		}
		return -1;
	}

	/** 返回新版本号 **/
	public static int setData(String path, String data) throws KeeperException, InterruptedException {
		return setData(path, data, -1);
	}

	/** 返回新版本号 **/
	public static int setData(String path, String data, int version) throws KeeperException, InterruptedException {
		return setData(path, data == null ? null : data.getBytes(), version);
	}

	/** 返回新版本号 **/
	public static int setData(String path, byte[] data, int version) throws KeeperException, InterruptedException {
		if (StringUtils.isEmpty(path)) {
			return -1;
		}

		checkConnect();

		if (null == zooKeeper.exists(path, false)) {
			createZNode(path);
		}
		Stat stat = zooKeeper.setData(path, data, version);
		if (null != stat) {
			if (LOGGER.isInfoEnabled()) {
				String string = null == data ? null : new String(data, 0, Math.min(100, SizeUtil.sizeZeroIfNull(data)));
				LOGGER.info(String.format("%s-设置'%s'='(%s)%s' 成功，version: '%s'", actionName, path, SizeUtil.size(data), string, stat.getVersion()));
			}
			return stat.getVersion();
		}
		return -1;
	}

	public static int setData(String path, Object obj) {
		checkConnect();

		try {
			return setData(path, JSON.toJSONString(obj));
		} catch (KeeperException | InterruptedException e) {
			LOGGER.error(String.format("%s-存入zookeeper -路径:  '%s'  -出错: '%s'", actionName, path, e.getMessage()), e);
		}
		return -1;
	}

	public static List<String> getChildren(String path) {
		checkConnect();

		if (StringUtils.isEmpty(path)) {
			return Collections.emptyList();
		}

		try {
			return zooKeeper.getChildren(path, false);
		} catch (Exception e) {
			LOGGER.error(String.format("%s-获取zk(%s)中的Children(%s)-出错: '%s'", actionName, zkConnectString, path, e.getMessage()), e);
		}
		return Collections.emptyList();
	}

	/**
	 * @param groupId
	 *            分组编号
	 * @param topic
	 *            主题
	 * @param newOffset
	 *            新的offset值
	 * @return 返回新值的版本号，-1表示出错
	 */
	public static int setKafkaOffset(String groupId, String topic, long newOffset) {
		if (StringUtils.isEmpty(groupId) || StringUtils.isEmpty(topic)) {
			return -1;
		}

		String path = toConsumerOffsetPath(groupId, topic);
		try {
			return setData(path, String.valueOf(newOffset));
		} catch (Exception e) {
			LOGGER.error(String.format("%s-更新zk(%s)中的Offset('%s'='%s')-出错: '%s'", actionName, zkConnectString, path, newOffset, e.getMessage()), e);
		}
		return -1;
	}

	/**
	 * @return 返回获取到的offset值，-1表示出错
	 */
	public static long getKafkaOffset(String groupId, String topic) {
		if (StringUtils.isEmpty(groupId) || StringUtils.isEmpty(topic)) {
			return -1;
		}

		String path = toConsumerOffsetPath(groupId, topic);
		try {
			String result = getData(path);
			return null != result ? Long.parseLong(result) : -2;
		} catch (Exception e) {
			LOGGER.error(String.format("%s-获取offset出错: '%s', path: '%s'", actionName, e.getMessage(), path), e);
		}
		return -1;
	}

	public static ProgramStatus getProgramStatus(String groupId) {
		if (StringUtils.isNotEmpty(groupId)) {
			String path = toProgramStatusPath(groupId);
			try {
				String result = getData(path);
				if (null != result) {
					return ProgramStatus.parse(result);
				}
			} catch (Exception e) {
				LOGGER.error(String.format("%s-获取programStatus(%s)出错: '%s'", actionName, path, e.getMessage()), e);
			}
		}
		return null;
	}

	public static void setProgramStatus(String groupId, ProgramStatus programStatus) {
		if (StringUtils.isNotEmpty(groupId)) {
			String path = toProgramStatusPath(groupId);
			try {
				setData(path, null == programStatus ? null : programStatus.toString());
			} catch (Exception e) {
				LOGGER.error(String.format("%s-设置zk(%s)中的programStatus('%s'='%s')-出错: '%s'", actionName, zkConnectString, path, programStatus, e.getMessage()), e);
			}
		}
	}

	public static ProgramStatus getLastProgramStatus(String groupId) {
		if (StringUtils.isNotEmpty(groupId)) {
			String path = toLastProgramStatusPath(groupId);
			try {
				String result = getData(path);
				if (null != result) {
					return ProgramStatus.parse(result);
				}
			} catch (Exception e) {
				LOGGER.error(String.format("%s-获取programStatus(%s)出错: '%s'", actionName, path, e.getMessage()), e);
			}
		}
		return null;
	}

	public static void setLastProgramStatus(String groupId, ProgramStatus programStatus) {
		if (StringUtils.isNotEmpty(groupId)) {
			String path = toLastProgramStatusPath(groupId);
			try {
				setData(path, null == programStatus ? null : programStatus.toString());
			} catch (Exception e) {
				LOGGER.error(String.format("%s-设置zk(%s)中的programStatus('%s'='%s')-出错: '%s'", actionName, zkConnectString, path, programStatus, e.getMessage()), e);
			}
		}
	}

	private static ZooKeeper checkConnect() {
		if (null == zkConnectString) {
			String message = String.format("%s-zk链接字符串为null，无法建立链接(需要提前设置)", actionName);
			LOGGER.error(message);
			throw new RuntimeException(message);
		}

		if (isAlive() == false) {
			synchronized (ZooKeeperUtil.class) {
				if (isAlive() == false) {
					cleanAndCreateConnect();
				}
			}
		}

		return zooKeeper;
	}

	private static void cleanAndCreateConnect() {
		if (null != zooKeeper) {
			try {
				zooKeeper.close();
				zooKeeper = null;
			} catch (InterruptedException e) {
			}
		}
		int maxTry = MAX_RETRY + 1;
		for (int i = 0; null == zooKeeper && i < maxTry; i++) {
			try {
				CountDownLatch connectedLatch = new CountDownLatch(1);
				Watcher watcher = new ConnectedWatcher(connectedLatch);
				zooKeeper = new ZooKeeper(zkConnectString, 3000, watcher);
				waitUntilConnected(zooKeeper, connectedLatch);
				break;
			} catch (IOException e) {
				LOGGER.error(String.format("%s-创建zk链接失败(剩余重试次数: '%s'): '%s'", actionName, maxTry - i - 1, e.getMessage()), e);
				try {
					Thread.sleep(1000);
				} catch (InterruptedException e1) {
				}
			}
		}
	}

	public static void waitUntilConnected(ZooKeeper zooKeeper, CountDownLatch connectedLatch) {
		if (States.CONNECTING == zooKeeper.getState()) {
			try {
				connectedLatch.await();
			} catch (InterruptedException e) {
				throw new IllegalStateException(e);
			}
		}
	}

	static class ConnectedWatcher implements Watcher {

		private CountDownLatch connectedLatch;

		ConnectedWatcher(CountDownLatch connectedLatch) {
			this.connectedLatch = connectedLatch;
		}

		@Override
		public void process(WatchedEvent event) {
			switch (event.getState()) {
			case SyncConnected:
				connectedLatch.countDown();
				break;
			case Expired:
				LOGGER.warn("{}-链接已过期({})，重新链接...", actionName, event.getState());
				cleanAndCreateConnect();
				break;
			default:
				LOGGER.warn("{}-状态已更改为: '{}'", actionName, event.getState());
				break;
			}
		}
	}

	private static boolean isAlive() {
		return null != zooKeeper && zooKeeper.getState().isConnected() && zooKeeper.getState().isAlive();
	}

	private static String toConsumerOffsetPath(String groupId, String topic) {
		int partition = 0;
		return String.format("/consumers/%s/offsets/%s/%s", groupId, topic, partition);
	}

	private static String toProgramStatusPath(String groupId) {
		return String.format("/inc/%s/programStatus", groupId);
	}

	private static String toLastProgramStatusPath(String groupId) {
		return String.format("/inc/%s/programStatus.last", groupId);
	}

	public static String toIncPath() {
		return "/inc";
	}

	public static String toIncIndexListPath(String groupId) {
		return String.format("/inc/%s/indexList", groupId);
	}

	public static String toIncLastIndexListPath(String groupId) {
		return String.format("/inc/%s/indexList.last", groupId);
	}

	public static String createZNode(String path) throws KeeperException, InterruptedException {
		return createZNode(path, CreateMode.PERSISTENT);
	}

	public static String createZNode(String path, CreateMode createMode) throws KeeperException, InterruptedException {
		if (StringUtils.isEmpty(path)) {
			return null;
		}

		checkConnect();

		StringBuilder sb = new StringBuilder();
		String[] split = path.split(ZK_PATH_SEPARATOR);
		String actualPath = null;
		for (int i = 0; i < split.length; i++) {
			makeZPath(sb, split[i]);
			if (null == zooKeeper.exists(sb.toString(), false)) {
				actualPath = zooKeeper.create(sb.toString(), null, Ids.OPEN_ACL_UNSAFE, i == split.length - 1 ? createMode : CreateMode.PERSISTENT);
				LOGGER.info(String.format("%s-创建节点: '%s'", actionName, actualPath));
			}
		}
		return actualPath;
	}

	private static StringBuilder makeZPath(StringBuilder sb, String string) {
		if (sb.length() <= 0) {
			sb.append(ZK_PATH_SEPARATOR);
		} else {
			if (ZK_PATH_SEPARATOR.equals(sb.substring(sb.length() - 1)) == false) {
				sb.append(ZK_PATH_SEPARATOR);
			}
		}
		sb.append(string);
		return sb;
	}

}
